package com.wtw.graph

import org.apache.spark.graphx.{Edge, EdgeTriplet, Graph}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object MapGraphXDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    //顶点的编号以及每个顶点中所含的属性
    val vertexArray: Array[(Long, (String, Int))] = Array(
      (1L, ("Alice", 28)),
      (2L, ("Bob", 27)),
      (3L, ("Charlie", 65)),
      (4L, ("David", 42)),
      (5L, ("Ed", 55)),
      (6L, ("Fran", 50))
    )

    val edgeArray: Array[Edge[Int]] = Array(
      Edge(2L, 1L, 7),
      Edge(2L, 4L, 2),
      Edge(3L, 2L, 4),
      Edge(3L, 6L, 3),
      Edge(4L, 1L, 1),
      Edge(5L, 2L, 2),
      Edge(5L, 3L, 8),
      Edge(5L, 6L, 3)
    )

    val vertexs: RDD[(Long, (String, Int))] = sc.parallelize(vertexArray)
    val edges: RDD[Edge[Int]] = sc.parallelize(edgeArray)

    //    （1）通过上面的项点数据和边数据创建图对象
    val graph = Graph(vertexs, edges)

    //    （2）使用mapEdges函数遍历所有的边，新增加一个属性值然后构建出新的图

    val graph2: Graph[(String, Int), String] = graph.mapEdges(edge => "边值: " + edge.attr)
    graph2.edges.collect().foreach(println(_))
    println("-------------使用mapEdges函数遍历所有的边，新增加一个属性值然后构建出新的图--------------")

    //    （3）使用mapTriplets函数遍历所有的三元组，新增加一个属性值，然后返回新的图

    //(String, Int)是顶点的属性，Int是边的属性
    //srcAttr, destAttr是边两个顶点的属性 attr是边的属性
    //mapVertices是对顶点的操作 用新的顶点属性创建成一个新图来返回
    //mapEdges/mapTriplets是对边的操作 用新的边属性创建成一个新图来返回
    val graph3: Graph[(String, Int), (String, Boolean)] = graph
      .mapTriplets((triplet: EdgeTriplet[(String, Int), Int]) => (triplet.attr + " wow", true))

    graph3.edges.collect().foreach(println(_))
    println("-------------使用mapTriplets函数遍历所有的三元组，新增加一个属性值，然后返回新的图--------------")
  }
}
